package com.parallel;
/**
 *
 **/

import lombok.extern.slf4j.Slf4j;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.util.StopWatch;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

/**
 * @Author weiwei
 * @Date 2021-03-03 16:14
 * @description 并行执行思考 摘自Java业务开发常见错误100例 31加餐1
 *
 **/
@Slf4j
public class ParallelThink {


    /**
     * 使用 20 个线程（threadCount）以并行方式总计执行 10000 次（taskCount）操作。
     * 因为单个任务单线程执行需要 10 毫秒（任务代码如下），也就是每秒吞吐量是 100 个操作，
     * 那 20 个线程 QPS 是 2000，执行完 10000 次操作最少耗时 5 秒。
     */
    @Test
    public  void invoke() throws Exception {
        int taskCount = 10000;
        int threadCount = 20;
        StopWatch stopWatch = new StopWatch();


        stopWatch.start("countDownLatch");
        Assert.assertEquals(taskCount,countDownLatchMethod(taskCount, threadCount));
        stopWatch.stop();

        stopWatch.start("executorService");
        Assert.assertEquals(taskCount,executorService(taskCount, threadCount));
        stopWatch.stop();

        /**
         * 个人认为下面三种方法本质的执行其实还是靠stream的并行流  只是依靠不同的方式进行堵塞
         **/
        stopWatch.start("forkJoinPool");
        Assert.assertEquals(taskCount,forkJoinPool(taskCount, threadCount));
        stopWatch.stop();

        stopWatch.start("parallel");
        Assert.assertEquals(taskCount,parallel(taskCount, threadCount));
        stopWatch.stop();

        stopWatch.start("completableFuture");
        Assert.assertEquals(taskCount,completableFuture(taskCount, threadCount));
        stopWatch.stop();

        log.info(stopWatch.prettyPrint());
    }

    private void increment(AtomicInteger atomicInteger){
        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        atomicInteger.incrementAndGet();
    }

    private int countDownLatchMethod(int taskCount, int threadCount) throws InterruptedException {
        AtomicInteger result = new AtomicInteger();
        CountDownLatch countDownLatch = new CountDownLatch(threadCount);
        IntStream.rangeClosed(1,threadCount).mapToObj(i->new Thread(()->{
            IntStream.rangeClosed(1,taskCount/threadCount).forEach(j->increment(result));
            countDownLatch.countDown();
        })).forEach(Thread::start);
        countDownLatch.await();
        return result.get();
    }


    /**
     *
     * @param taskCount
     * @param threadCount
     * @return
     * @throws InterruptedException
     * shutdown()：停止接收新任务，原来的任务继续执行
     * shutdownNow()：停止接收新任务，原来的任务停止执行(使用interrupt中断)
     * awaitTermination(long timeOut, TimeUnit unit)：当前线程阻塞,等待线程池执行完或者时间到了,或者线程被中断，抛出InterruptedException  可以继续向线程池提价任务
     */
    private int executorService(int taskCount, int threadCount) throws InterruptedException {
        AtomicInteger result = new AtomicInteger();
        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
        IntStream.rangeClosed(1,taskCount).forEach(i->executorService.execute(()->increment(result)));
        executorService.shutdown();
        executorService.awaitTermination(1,TimeUnit.HOURS);
        return result.get();
    }

    //forkJoinPooln n并行度有n个独立队列
    //ForkJoinPool 更适合大任务分割成许多小任务并行执行的场景，而 ThreadPoolExecutor 适合许多独立任务并发执行的场景
    private int forkJoinPool(int taskCount, int threadCount) throws InterruptedException {
        AtomicInteger result = new AtomicInteger();
        ForkJoinPool forkJoinPool=new ForkJoinPool(taskCount);
        forkJoinPool.execute(()->IntStream.rangeClosed(1,taskCount).parallel().forEach(i->increment(result)));
        forkJoinPool.shutdown();
        forkJoinPool.awaitTermination(1,TimeUnit.HOURS);
        return result.get();
    }

    /**
     * 直接使用并行流，并行流使用公共的 ForkJoinPool
     * 公共的 ForkJoinPool 默认的并行度是 CPU 核心数 -1
     *
     * @param taskCount
     * @param threadCount
     * @return
     */
    private int parallel(int taskCount, int threadCount){
        //设置公共ForkJoinPool的并行度
        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", String.valueOf(threadCount));
        AtomicInteger result = new AtomicInteger();
        IntStream.rangeClosed(1,taskCount).parallel().forEach(i->increment(result));
        return result.get();
    }

    /**
     * CompletableFuture 来实现。
     * CompletableFuture.runAsync 方法可以指定一个线程池，一般会在使用 CompletableFuture
     * @param taskCount
     * @param threadCount
     * @return
     * @throws ExecutionException
     * @throws InterruptedException
     */
    private int completableFuture(int taskCount, int threadCount) throws ExecutionException, InterruptedException {
       AtomicInteger result = new AtomicInteger();
        ForkJoinPool forkJoinPool = new ForkJoinPool(threadCount);
        CompletableFuture.runAsync(()->IntStream.rangeClosed(1,taskCount).parallel().forEach(i->increment(result)),forkJoinPool).get();
        return result.get();
    }
}
